package com.xujian.flink.split;

import com.xujian.common.beans.EventSplit;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author star xu
 * @date 2023/1/4 15:23
 * @Description: 分流
 * @Slogan: 致敬大师,致敬未来的自己
 */
public class SplitStreamByOutputTag {
    private static final OutputTag<Tuple3<String, String, Long>> MARY_TAG = new OutputTag<Tuple3<String, String, Long>>("令狐冲") {};
    private static final OutputTag<Tuple3<String, String, Long>> BOB_TAG = new OutputTag<Tuple3<String, String, Long>>("风清扬") {};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<EventSplit> stream = env.addSource(new MySource());
        SingleOutputStreamOperator<EventSplit> processedStream = stream.process(new ProcessFunction<EventSplit, EventSplit>() {
            @Override
            public void processElement(EventSplit value, Context ctx, Collector<EventSplit> out) throws Exception {
                if ("令狐冲".equals(value.user)) {
                    ctx.output(MARY_TAG, new Tuple3<>(value.user, value.url, value.timeStamp));
                } else if ("风清扬".equals(value.user)) {
                    ctx.output(BOB_TAG, new Tuple3<>(value.user, value.url, value.timeStamp));
                } else {
                    out.collect(value);
                }
            }
        });

        processedStream.getSideOutput(MARY_TAG).print("令狐冲");
        processedStream.getSideOutput(BOB_TAG).print("风清扬");
        processedStream.print("else");

        env.execute();
    }
}
